package com.wugui.datatx.core.biz.impl;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.wugui.datatx.core.biz.ExecutorBiz;
import com.wugui.datatx.core.biz.model.LogResult;
import com.wugui.datatx.core.biz.model.ReturnT;
import com.wugui.datatx.core.biz.model.TriggerParam;
import com.wugui.datatx.core.enums.ExecutorBlockStrategyEnum;
import com.wugui.datatx.core.executor.JobExecutor;
import com.wugui.datatx.core.glue.GlueFactory;
import com.wugui.datatx.core.glue.GlueTypeEnum;
import com.wugui.datatx.core.handler.IJobHandler;
import com.wugui.datatx.core.handler.impl.GlueJobHandler;
import com.wugui.datatx.core.handler.impl.ScriptJobHandler;
import com.wugui.datatx.core.log.JobFileAppender;
import com.wugui.datatx.core.thread.JobThread;

/**
 * Created by xuxueli on 17/3/1.
 */

public class ExecutorBizImpl implements ExecutorBiz {
	private static Logger logger = LoggerFactory.getLogger(ExecutorBizImpl.class);

	@Override
	public ReturnT<String> beat() {
		return ReturnT.SUCCESS;
	}

	@Override
	public ReturnT<String> idleBeat(int jobId) {

		// isRunningOrHasQueue
		JobThread jobThread = JobExecutor.loadJobThread(jobId);
		if (jobThread != null && jobThread.isRunningOrHasQueue()) {
			return new ReturnT<>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue.");
		}
		return ReturnT.SUCCESS;
	}

	@Override
	public ReturnT<String> kill(int jobId) {
		// kill handlerThread, and create new one
		JobThread jobThread = JobExecutor.loadJobThread(jobId);
		if (jobThread != null) {
			JobExecutor.removeJobThread(jobId, "scheduling center kill job.");
			return ReturnT.SUCCESS;
		}

		return new ReturnT<>(ReturnT.SUCCESS_CODE, "job thread already killed.");
	}

	@Override
	public ReturnT<LogResult> log(long logDateTim, long logId, int fromLineNum) {
		// log filename: logPath/yyyy-MM-dd/9999.log
		String logFileName = JobFileAppender.makeLogFileName(new Date(logDateTim), logId);

		LogResult logResult = JobFileAppender.readLog(logFileName, fromLineNum);
		return new ReturnT<>(logResult);
	}

	@Override
	public ReturnT<String> run(TriggerParam triggerParam) {
		// load old：jobHandler + jobThread
		JobThread jobThread = JobExecutor.loadJobThread(triggerParam.getJobId());
		IJobHandler jobHandler = jobThread != null ? jobThread.getHandler() : null;
		String removeOldReason = null;

		// valid：jobHandler + jobThread
		GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType());
		if (GlueTypeEnum.DATAX == glueTypeEnum || GlueTypeEnum.BEAN == glueTypeEnum) {

			// new jobhandler
			IJobHandler newJobHandler = JobExecutor.loadJobHandler(triggerParam.getExecutorHandler());

			// valid old jobThread
			if (jobThread != null && jobHandler != newJobHandler) {
				// change handler, need kill old thread
				removeOldReason = "change jobhandler or glue type, and terminate the old job thread.";

				jobThread = null;
				jobHandler = null;
			}

			// valid handler
			if (jobHandler == null) {
				jobHandler = newJobHandler;
				if (jobHandler == null) {
					return new ReturnT<>(ReturnT.FAIL_CODE,
							"job handler [" + triggerParam.getExecutorHandler() + "] not found.");
				}
			}

		} else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) {

			// valid old jobThread
			if (jobThread != null
					&& !(jobThread.getHandler() instanceof GlueJobHandler && ((GlueJobHandler) jobThread.getHandler())
							.getGlueUpdatetime() == triggerParam.getGlueUpdatetime())) {
				// change handler or gluesource updated, need kill old thread
				removeOldReason = "change job source or glue type, and terminate the old job thread.";

				jobThread = null;
				jobHandler = null;
			}

			// valid handler
			if (jobHandler == null) {
				try {
					IJobHandler originJobHandler = GlueFactory.getInstance()
							.loadNewInstance(triggerParam.getGlueSource());
					jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime());
				} catch (Exception e) {
					logger.error(e.getMessage(), e);
					return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
				}
			}
		} else if (glueTypeEnum != null && glueTypeEnum.isScript()) {

			// valid old jobThread
			if (jobThread != null && !(jobThread.getHandler() instanceof ScriptJobHandler
					&& ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime() == triggerParam
							.getGlueUpdatetime())) {
				// change script or gluesource updated, need kill old thread
				removeOldReason = "change job source or glue type, and terminate the old job thread.";

				jobThread = null;
				jobHandler = null;
			}

			// valid handler
			if (jobHandler == null) {
				jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(),
						triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType()));
			}
		} else {
			return new ReturnT<>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
		}

		// executor block strategy
		if (jobThread != null) {
			ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum
					.match(triggerParam.getExecutorBlockStrategy(), null);
			if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
				// discard when running
				if (jobThread.isRunningOrHasQueue()) {
					return new ReturnT<>(ReturnT.FAIL_CODE,
							"block strategy effect：" + ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
				}
			} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
				// kill running jobThread
				if (jobThread.isRunningOrHasQueue()) {
					removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();

					jobThread = null;
				}
			} else {
				// just queue trigger
			}
		}

		// replace thread (new or exists invalid)
		if (jobThread == null) {
			jobThread = JobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
		}

		// push data to queue
		ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);
		return pushResult;
	}

}
